package com.atguigu.sink;

import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MySink extends AbstractSink implements Configurable {
    private String prefix;
    private String suffix;
    private static final Logger logger = LoggerFactory.getLogger(AbstractSink.class);
    @Override
    public Status process() throws EventDeliveryException {

        //声明返回值状态
        Status status;
        //获取channel
        Channel ch = getChannel();
        //获取事务
        Transaction txn = ch.getTransaction();
        //声明事件
        Event event;
        //开启事务
        txn.begin();
        //读取事件
        while(true) {
            event = ch.take();
            if (event != null) {
                break;
            }
        }
        try {
            //处理事件(打印)
            logger.info(prefix+new String(event.getBody())+suffix);
            //事务提交
            txn.commit();
            status=Status.READY;
        } catch (Exception e) {
            e.printStackTrace();
            txn.rollback();
            status=Status.BACKOFF;
        } finally {
            txn.close();
        }
        return status;
    }

    @Override
    public void configure(Context context) {
        prefix=context.getString("prefix");
        suffix = context.getString("suffix");
    }
}
